#include <ddm/AtomicFib.h>
#include <ddm/AtomicRoute.h>
#include <ddm/Config.h>
#include <ddm/MapReducer.h>
#include <ddm/OpenrAgent.h>
#include <ddm/ThriftServer.h>
#include <folly/init/Init.h>
#include <folly/logging/xlog.h>
#include <openr/common/OpenrEventBase.h>
#include <openr/watchdog/Watchdog.h>
#include <syslog.h>

#include <iostream>

// The node name can be fetched from OpenR
DEFINE_string(node_name, "dummy_node", "Current node name");
DEFINE_string(host, "::1", "Host to connect to");
DEFINE_int32(port, openr::Constants::kOpenrCtrlPort, "OpenrCtrl server port");
DEFINE_int32(connect_timeout_ms, 1000, "Connect timeout for client");
DEFINE_int32(processing_timeout_ms, 5000, "Processing timeout for client");

/**
 * Start an EventBase in a thread, maintain order of thread creation and
 * returns raw pointer of Derived class.
 */
template <typename T>
T* startEventBase(
    std::vector<std::thread>& allThreads,
    std::vector<std::unique_ptr<openr::OpenrEventBase>>& orderedEvbs,
    openr::Watchdog* watchdog, const std::string& name,
    std::unique_ptr<T> evbT) {
  CHECK(evbT);
  auto t = evbT.get();
  auto evb = std::unique_ptr<openr::OpenrEventBase>(
      reinterpret_cast<openr::OpenrEventBase*>(evbT.release()));
  evb->setEvbName(name);

  // Start a thread
  allThreads.emplace_back(std::thread([evb = evb.get(), name]() noexcept {
    XLOG(INFO) << "Starting " << name << " thread ...";
    folly::setThreadName(fmt::format("ddm-{}", name));
    evb->run();
    XLOG(INFO) << name << " thread got stopped.";
  }));
  evb->waitUntilRunning();

  // Add to watchdog
  if (watchdog) {
    watchdog->addEvb(evb.get());
  }

  // Emplace evb into ordered list of evbs. So that we can destroy
  // them in revserse order of their creation.
  orderedEvbs.emplace_back(std::move(evb));

  return t;
}

int main(int argc, char** argv) {
  std::cout << "Greetings!" << std::endl;
  // Initialize all params
  folly::init(&argc, &argv);

  // Register the signals to handle before anything else. This guarantees that
  // any threads created below will inherit the signal mask
  folly::EventBase mainEvb;
  openr::EventBaseStopSignalHandler handler(&mainEvb);
  folly::setThreadName("ddm");

  std::shared_ptr<ddm::Config> config;

  openr::messaging::ReplicateQueue<openr::DecisionRouteUpdate> ddmRouteUpdatesQueue;
  openr::messaging::ReplicateQueue<ddm::AtomicRoute> atomicRouteUpdatesQueue;
  openr::messaging::ReplicateQueue<ddm::thrift::Requirement> requirementsQueue;

  // structures to organize our modules
  std::vector<std::thread> allThreads;
  std::vector<std::unique_ptr<openr::OpenrEventBase>> orderedEvbs;
  openr::Watchdog* watchdog{nullptr};

  // Starting main event-loop
  std::thread mainEvbThread([&]() {
    XLOG(INFO) << "Starting ddm main event-base...";
    folly::setThreadName("ddm-main");
    mainEvb.loopForever();
    XLOG(INFO) << "Main event-base stopped...";
  });
  mainEvb.waitUntilRunning();

  auto mapReducer = startEventBase(
      allThreads, orderedEvbs, watchdog, "mapReducer",
      std::make_unique<ddm::MapReducer>(atomicRouteUpdatesQueue.getReader(),
                                        requirementsQueue.getReader(),
                                        FLAGS_node_name));

  auto atomicFib =
      startEventBase(allThreads, orderedEvbs, watchdog, "atomicFib",
                     std::make_unique<ddm::AtomicFib>(
                         std::move(ddmRouteUpdatesQueue.getReader()),
                         atomicRouteUpdatesQueue));

  // Start openrAgent Module
  auto openrAgent =
      startEventBase(allThreads, orderedEvbs, watchdog, "openrAgent",
                     std::make_unique<ddm::OpenrAgent>(FLAGS_node_name, config, requirementsQueue));
  openrAgent->startSubscribe();

  // Start ddm control server
  auto ddmCtrlOpenrEvb = std::make_unique<openr::OpenrEventBase>();
  auto ddmCtrlHandler = std::make_shared<ddm::CtrlHandler>(
      ddmRouteUpdatesQueue, ddmCtrlOpenrEvb.get(), FLAGS_node_name);
  startEventBase(allThreads, orderedEvbs, watchdog, "ddm_ctrl_evb",
                 std::move(ddmCtrlOpenrEvb));

  auto ddmThriftServer = std::make_unique<ddm::ThriftServer>(ddmCtrlHandler);
  ddmThriftServer->start();

  // Wait for main eventbase to stop
  XLOG(INFO) << "main thread join";
  mainEvbThread.join();

  // Close all queues
  ddmRouteUpdatesQueue.close();
  atomicRouteUpdatesQueue.close();
  requirementsQueue.close();

  // Stop & destroy thrift server. Will reduce ref-count on ctrlHandler
  ddmThriftServer->stop();
  ddmThriftServer.reset();
  // Destroy ctrlHandler
  CHECK(ddmCtrlHandler.unique()) << "Unexpected ownership of ctrlHandler pointer";
  ddmCtrlHandler.reset();

  // Stop all threads (in reverse order of their creation)
  for (auto riter = orderedEvbs.rbegin(); orderedEvbs.rend() != riter;
       ++riter) {
    XLOG(INFO) << "Stopping " << (*riter)->getEvbName();
    (*riter)->stop();
    (*riter)->waitUntilStopped();
    XLOG(INFO) << "Finally stopped " << (*riter)->getEvbName();
  }

  // Wait for all threads
  for (auto& t : allThreads) {
    t.join();
  }
  orderedEvbs.clear();

  // Close syslog connection (this is optional)
  SYSLOG(INFO) << "Stopping DDM daemon: ppid = " << getpid();
  closelog();

  return 0;
}